package com.powerbank.common.mq;

import com.powerbank.common.event.BaseEvent;
import com.powerbank.common.exception.BusinessException;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 事件发布器
 * 统一管理所有业务事件的发布
 */
@Service
public class EventPublisher {
    
    private static final Logger log = LoggerFactory.getLogger(EventPublisher.class);
    
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    
    // Topic常量
    public static final String USER_TOPIC = "PowerBank-User-Topic";
    public static final String PAYMENT_TOPIC = "PowerBank-Payment-Topic";
    public static final String ORDER_TOPIC = "PowerBank-Order-Topic";
    public static final String DEVICE_TOPIC = "PowerBank-Device-Topic";
    public static final String FINANCE_TOPIC = "PowerBank-Finance-Topic";
    public static final String NOTIFICATION_TOPIC = "PowerBank-Notification-Topic";
    
    // 重试相关配置
    private static final int MAX_RETRY_COUNT = 3;
    private final ConcurrentHashMap<String, AtomicInteger> retryCountMap = new ConcurrentHashMap<>();
    
    /**
     * 发布用户相关事件
     *
     * @param event 事件对象
     */
    public void publishUserEvent(BaseEvent event) {
        validateEvent(event);
        publishEvent(USER_TOPIC, event);
    }
    
    /**
     * 发布支付相关事件
     *
     * @param event 事件对象
     */
    public void publishPaymentEvent(BaseEvent event) {
        validateEvent(event);
        publishEvent(PAYMENT_TOPIC, event);
    }
    
    /**
     * 发布订单相关事件
     *
     * @param event 事件对象
     */
    public void publishOrderEvent(BaseEvent event) {
        validateEvent(event);
        publishEvent(ORDER_TOPIC, event);
    }
    
    /**
     * 发布设备相关事件
     *
     * @param event 事件对象
     */
    public void publishDeviceEvent(BaseEvent event) {
        validateEvent(event);
        publishEvent(DEVICE_TOPIC, event);
    }
    
    /**
     * 发布财务相关事件
     *
     * @param event 事件对象
     */
    public void publishFinanceEvent(BaseEvent event) {
        validateEvent(event);
        publishEvent(FINANCE_TOPIC, event);
    }
    
    /**
     * 发布通知相关事件
     *
     * @param event 事件对象
     */
    public void publishNotificationEvent(BaseEvent event) {
        validateEvent(event);
        publishEvent(NOTIFICATION_TOPIC, event);
    }
    
    /**
     * 发布事务消息（用于需要保证事务一致性的场景）
     *
     * @param topic  主题
     * @param event  事件对象
     * @param arg    事务参数
     */
    public void publishTransactionEvent(String topic, BaseEvent event, Object arg) {
        // 参数校验
        if (!StringUtils.hasText(topic)) {
            throw new BusinessException("主题不能为空");
        }
        validateEvent(event);
        
        try {
            String destination = topic + ":" + event.getEventType();
            Message<BaseEvent> message = MessageBuilder
                    .withPayload(event)
                    .setHeader("eventId", event.getEventId())
                    .setHeader("eventType", event.getEventType())
                    .setHeader("sourceService", event.getSourceService())
                    .build();
            
            log.info("发布事务消息: topic={}, eventType={}, eventId={}", 
                    topic, event.getEventType(), event.getEventId());
            
            rocketMQTemplate.sendMessageInTransaction(destination, message, arg);
            
            log.info("事务消息发布成功: topic={}, eventType={}, eventId={}", 
                    topic, event.getEventType(), event.getEventId());
            
            // 发布成功，清除重试计数
            retryCountMap.remove(event.getEventId());
        } catch (Exception e) {
            log.error("发布事务消息失败: topic={}, eventType={}, eventId={}", 
                    topic, event.getEventType(), event.getEventId(), e);
            throw new BusinessException("事务消息发布失败: " + e.getMessage());
        }
    }
    
    /**
     * 通用事件发布方法
     *
     * @param topic 主题
     * @param event 事件对象
     */
    private void publishEvent(String topic, BaseEvent event) {
        try {
            String destination = topic + ":" + event.getEventType();
            
            log.info("发布事件: topic={}, eventType={}, eventId={}", 
                    topic, event.getEventType(), event.getEventId());
            
            rocketMQTemplate.syncSend(destination, event);
            
            log.info("事件发布成功: topic={}, eventType={}, eventId={}", 
                    topic, event.getEventType(), event.getEventId());
            
            // 发布成功，清除重试计数
            retryCountMap.remove(event.getEventId());
        } catch (Exception e) {
            log.error("事件发布失败: topic={}, eventType={}, eventId={}", 
                    topic, event.getEventType(), event.getEventId(), e);
            
            // 尝试重试
            if (shouldRetry(event.getEventId())) {
                log.info("尝试重试发布事件: topic={}, eventType={}, eventId={}", 
                        topic, event.getEventType(), event.getEventId());
                retryPublishEvent(topic, event);
            } else {
                log.error("事件发布重试次数已达到上限: topic={}, eventType={}, eventId={}", 
                        topic, event.getEventType(), event.getEventId());
                // 可以根据业务需求决定是否将失败的事件记录到数据库或错误队列
            }
        }
    }
    
    /**
     * 异步发布事件
     *
     * @param topic 主题
     * @param event 事件对象
     */
    public void publishEventAsync(String topic, BaseEvent event) {
        // 参数校验
        if (!StringUtils.hasText(topic)) {
            log.warn("异步发布事件失败: 主题不能为空");
            return;
        }
        validateEvent(event);
        
        try {
            String destination = topic + ":" + event.getEventType();
            
            log.info("异步发布事件: topic={}, eventType={}, eventId={}", 
                    topic, event.getEventType(), event.getEventId());
            
            rocketMQTemplate.asyncSend(destination, event, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    log.info("异步事件发布成功: topic={}, eventType={}, eventId={}, msgId={}", 
                            topic, event.getEventType(), event.getEventId(), sendResult.getMsgId());
                    // 发布成功，清除重试计数
                    retryCountMap.remove(event.getEventId());
                }
                
                @Override
                public void onException(Throwable e) {
                    log.error("异步事件发布失败: topic={}, eventType={}, eventId={}", 
                            topic, event.getEventType(), event.getEventId(), e);
                    
                    // 尝试重试
                    if (shouldRetry(event.getEventId())) {
                        log.info("尝试重试异步发布事件: topic={}, eventType={}, eventId={}", 
                                topic, event.getEventType(), event.getEventId());
                        retryPublishEventAsync(topic, event);
                    } else {
                        log.error("异步事件发布重试次数已达到上限: topic={}, eventType={}, eventId={}", 
                                topic, event.getEventType(), event.getEventId());
                        // 可以将失败的事件记录到数据库或错误队列，以便后续重试
                    }
                }
            });
            
        } catch (Exception e) {
            log.error("异步事件发布失败: topic={}, eventType={}, eventId={}", 
                    topic, event.getEventType(), event.getEventId(), e);
        }
    }
    
    /**
     * 延时发布事件
     *
     * @param topic       主题
     * @param event       事件对象
     * @param delayLevel  延时级别
     */
    public void publishDelayEvent(String topic, BaseEvent event, int delayLevel) {
        // 参数校验
        if (!StringUtils.hasText(topic)) {
            throw new BusinessException("主题不能为空");
        }
        validateEvent(event);
        
        // 验证延时级别
        if (delayLevel < 1 || delayLevel > 18) {
            log.warn("延时级别超出范围[1-18]，使用默认级别1");
            delayLevel = 1;
        }
        
        try {
            String destination = topic + ":" + event.getEventType();
            Message<BaseEvent> message = MessageBuilder
                    .withPayload(event)
                    .setHeader("eventId", event.getEventId())
                    .setHeader("eventType", event.getEventType())
                    .setHeader("sourceService", event.getSourceService())
                    .build();
            
            log.info("发布延时事件: topic={}, eventType={}, eventId={}, delayLevel={}", 
                    topic, event.getEventType(), event.getEventId(), delayLevel);
            
            rocketMQTemplate.syncSend(destination, message, 3000, delayLevel);
            
            log.info("延时事件发布成功: topic={}, eventType={}, eventId={}", 
                    topic, event.getEventType(), event.getEventId());
            
            // 发布成功，清除重试计数
            retryCountMap.remove(event.getEventId());
        } catch (Exception e) {
            log.error("延时事件发布失败: topic={}, eventType={}, eventId={}", 
                    topic, event.getEventType(), event.getEventId(), e);
            throw new BusinessException("延时事件发布失败: " + e.getMessage());
        }
    }
    
    /**
     * 验证事件对象
     *
     * @param event 事件对象
     */
    private void validateEvent(BaseEvent event) {
        if (event == null) {
            throw new BusinessException("事件对象不能为空");
        }
        
        if (!StringUtils.hasText(event.getEventType())) {
            throw new BusinessException("事件类型不能为空");
        }
        
        if (!StringUtils.hasText(event.getSourceService())) {
            throw new BusinessException("来源服务不能为空");
        }
        
        if (!StringUtils.hasText(event.getEventId())) {
            throw new BusinessException("事件ID不能为空");
        }
    }
    
    /**
     * 判断是否应该重试
     * 
     * @param eventId 事件ID
     * @return 是否应该重试
     */
    private boolean shouldRetry(String eventId) {
        AtomicInteger retryCount = retryCountMap.computeIfAbsent(eventId, k -> new AtomicInteger(0));
        return retryCount.get() < MAX_RETRY_COUNT;
    }
    
    /**
     * 重试发布事件
     * 
     * @param topic 主题
     * @param event 事件对象
     */
    private void retryPublishEvent(String topic, BaseEvent event) {
        AtomicInteger retryCount = retryCountMap.computeIfAbsent(event.getEventId(), k -> new AtomicInteger(0));
        retryCount.incrementAndGet();
        
        // 简单的延迟重试
        try {
            Thread.sleep(1000 * retryCount.get()); // 递增延迟
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        
        publishEvent(topic, event);
    }
    
    /**
     * 重试异步发布事件
     * 
     * @param topic 主题
     * @param event 事件对象
     */
    private void retryPublishEventAsync(String topic, BaseEvent event) {
        AtomicInteger retryCount = retryCountMap.computeIfAbsent(event.getEventId(), k -> new AtomicInteger(0));
        retryCount.incrementAndGet();
        
        // 简单的延迟重试
        try {
            Thread.sleep(1000 * retryCount.get()); // 递增延迟
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        
        publishEventAsync(topic, event);
    }
}